package com.nami.dispatcher.domain.service.impl;

import cn.hutool.core.collection.CollUtil;
import com.google.protobuf.ByteString;
import com.nami.dispatcher.domain.message.model.PacketTrans;
import com.nami.dispatcher.domain.middleware.mq.DomainEventPublisher;
import com.nami.dispatcher.domain.model.NtfRecvModel;
import com.nami.dispatcher.domain.qos.QosSendDaemonS2C;
import com.nami.dispatcher.domain.service.DispatchPrivateDomainService;
import com.nami.dispatcher.types.utils.IndexGenerator;
import com.nami.gate.external.basic.bo.ConnectedUserBO;
import com.nami.gate.external.basic.constants.DownTopicConst;
import com.nami.im.network.model.enums.protocol.CMDEnum;
import com.nami.im.network.model.message.v2.Message;
import com.nami.transport.proto.codec.ImMessages;
import com.nami.transport.proto.codec.MqTransMsg;
import com.nami.transport.proto.codec.UpWsMsg;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.dromara.common.core.exception.ServiceException;
import org.dromara.common.core.utils.SessionUtil;
import org.jetbrains.annotations.NotNull;
import org.springframework.stereotype.Service;


@Slf4j
@Service
@RequiredArgsConstructor
public class DispatchPrivateDomainServiceImpl implements DispatchPrivateDomainService {

    private final DomainEventPublisher mqSender;

    public void notifyReceiver(NtfRecvModel receiver) {
        Message message = receiver.getMessage();
        switch (message.getChannelType()) {
            case ImMessages.ChannelType.Private_VALUE:
                forwardMsg2Recv(receiver);
                break;
            case ImMessages.ChannelType.Group_VALUE:
                // 其他群成员
//                forwardMsg2GrpMemberClients(receiver);
                break;
            default:
                log.error("[dispatch] 坏了，不知道转哪了. msgId={}", message.getMsgId());
                throw new ServiceException("发送消息失败，. msgId=" + message.getMsgId());
        }
    }
    private ImMessages.DownMsg toDownMsg(Message message) {
        ImMessages.DownMsg downMsg = ImMessages.DownMsg.newBuilder()
            .setClientUid(message.getClientUid())
            .setMsgId(message.getMsgId())
            .setMsgContent(ByteString.copyFromUtf8(message.getMsgContent()))
            .setMsgTime(message.getMsgTime())
            .setPlatform(message.getPlatform())
            .setChannelType(ImMessages.ChannelType.Private)
            .setSenderId(message.getSenderId())
            .setTargetId(message.getTargetId())
            .setMsgSeqNo(message.getMsgSeqNo())
            .setMsgType(message.getMsgType())
            .build();
        return downMsg;
    }
    private void forwardMsg2Recv(NtfRecvModel ntfRecv) {
        ConnectedUserBO receiver = ntfRecv.getReceiver();
        Message message = ntfRecv.getMessage();
        ImMessages.DownMsg downMsg = toDownMsg(message);
        // 用户在线发送消息
        if (CollUtil.isNotEmpty(ntfRecv.getDevices())) {
            log.info("[c2c] 开始向用户【{}】的在线客户端, 转发消息！", receiver.getNick());
            ntfRecv.getDevices().stream().forEach(item -> {
                UpWsMsg.PublishMsg publishMsg = UpWsMsg.PublishMsg.newBuilder()
                    .setIndex(IndexGenerator.nextIndex(message.getSenderId(), message.getTargetId(), message.getMsgId()))
                    .setTopic("msg")
                    .setTargetId(ntfRecv.getSenderId())
                    .setData(downMsg.toByteString())
                    .build();

                String sessionId = SessionUtil.binding(receiver.getUserId(), item.getInstanceId());
                MqTransMsg.MessageHeader header = MqTransMsg.MessageHeader.newBuilder()
                    .setCmd(ByteString.copyFrom(new byte[]{CMDEnum.PUBLISH.getCode()}))
                    .setQos(ByteString.copyFrom(new byte[]{0}))
                    .setVersion(ByteString.copyFrom(new byte[]{0}))
                    .build();
                MqTransMsg.Dispatch2GatePublishResponse response = MqTransMsg.Dispatch2GatePublishResponse.newBuilder()
                    .setSessionId(sessionId)
                    .setMsgHeader(header)
                    .setDownMsg(publishMsg.toByteString())
                    .setTimestamp(System.currentTimeMillis())
                    .build();
                log.info("[c2c] send msg client:{}, serverMsgId:{}", sessionId, message.getMsgId());

                mqSender.publish(DownTopicConst.FLUSH_ONE_TOPIC, item.getConnectId(), response.toByteArray());
//                    ForwardPacketUtil.flushPacket(request, item.getDubboIp(), item.getDubboPort());
                if (ntfRecv.getQos() > 0) {
                    if (null != message.getMsgId() && !QosSendDaemonS2C.getInstance().exist(message.getMsgId())) {
                        QosSendDaemonS2C.getInstance().put(genQosTransBo(message.getMsgId(), response.toByteArray(), receiver.getAppKey(),
                            sessionId, item.getConnectId()));
                    }
                }

            });
            return;
        }
        log.info("[c2c] 用户不在线, 已保存离线消息。id:{}:{}", message.getMsgId(), message.getMsgContent());
    }

    @NotNull
    private PacketTrans genQosTransBo(String msgId, byte[] packet, String appkey, String sessionId, String mqTags) {
        PacketTrans packetTransBO = new PacketTrans();
        packetTransBO.setPacket(packet);
        packetTransBO.setQos(true);
        packetTransBO.setSessionId(sessionId);
        packetTransBO.setMessageId(msgId);
        packetTransBO.setAppKey(appkey);

        packetTransBO.setMqTags(mqTags);
        packetTransBO.setIfGroupMsg(false);
        return packetTransBO;
    }


}
